package com.wish233.wiliwilivideo.mq;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.toolkit.Db;
import com.github.benmanes.caffeine.cache.Cache;
import com.rabbitmq.client.Channel;
import com.wish233.wiliwilivideo.domain.dto.comment.CommentDeleteMessage;
import com.wish233.wiliwilivideo.domain.dto.comment.CommentSelectMessage;
import com.wish233.wiliwilivideo.domain.dto.likes.LikeCountMessage;
import com.wish233.wiliwilivideo.domain.dto.likes.LikeLocal2Redis;
import com.wish233.wiliwilivideo.domain.dto.likes.LikeOperationMessage;
import com.wish233.wiliwilivideo.domain.po.Likes;
import com.wish233.wiliwilivideo.domain.vo.CommentVO;
import com.wish233.wiliwilivideo.mapper.CommentMapper;
import com.wish233.wiliwilivideo.mapper.LikeCountsMapper;
import com.wish233.wiliwilivideo.mapper.LikesMapper;
import com.wish233.wiliwilivideo.mapper.VideoMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.wish233.wiliwilicommon.constants.MqConstant.DLX_EXCHANGE;
import static com.wish233.wiliwilicommon.constants.MqConstant.DLX_ROUTING_KEY;
import static com.wish233.wiliwilicommon.constants.RedisConstants.*;
import static com.wish233.wiliwilivideo.domain.constant.AMQPConstant.*;
import static com.wish233.wiliwilivideo.domain.constant.BizConstant.COMMENT_BIZ;
import static com.wish233.wiliwilivideo.domain.constant.BizConstant.VIDEO_BIZ;
import static com.wish233.wiliwilivideo.utils.RedisUtils.calculateHotScore;


/**
 * @author WHH
 */
@Service
@Slf4j
public class LikeMessageListener {

    @Resource
    private LikesMapper likesMapper;
    @Resource
    private LikeCountsMapper likeCountsMapper;
    @Resource
    private VideoMapper videoMapper;
    @Resource
    private CommentMapper commentMapper;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private MqProducer mqProducer;
    @Resource
    private Cache<String, List<CommentVO>> localCache;

    @SneakyThrows
    @Transactional(rollbackFor = RuntimeException.class, propagation = Propagation.REQUIRES_NEW)
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(name = LIKE_OPERATION_QUEUE, durable = "true", arguments = {
                            @Argument(name = "x-dead-letter-exchange", value = DLX_EXCHANGE), // 设置死信交换机
                            @Argument(name = "x-dead-letter-routing-key", value = DLX_ROUTING_KEY),// 设置死信路由键
                            @Argument(name = "x-message-ttl", value = "300000", type = "java.lang.Long")

                    }),
                    exchange = @Exchange(name = LIKE_OPERATION_EXCHANGE),
                    key = LIKE_OPERATION_ROUTING_KEY),
            ackMode = "MANUAL")
    public void handleLikeOpMessage(LikeOperationMessage message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            //拿到消息
            String operationType = message.getOperationType();
            Long userId = message.getUserId();
            Integer bizId = message.getBizId();
            Long messageId = message.getMessageId();
            String uuid = message.getId();
            Boolean absent = stringRedisTemplate.opsForValue().setIfAbsent(LIKE_MQ_ID_KEY + uuid, "ok", LIKE_MQ_ID_TTL, TimeUnit.MINUTES);
            //如果存在这个key,说明已经在处理了，返回ack
            if (Boolean.FALSE.equals(absent)) {
                channel.basicAck(deliveryTag, false);
            }
            //判断消息类型(是删除还是增加)
            QueryWrapper<Likes> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("user_id", userId)
                    .eq("biz_id", bizId)
                    .eq("message_id", messageId);
            if ("ADD".equals(operationType)) {
                // 幂等性检查
                Likes existing = likesMapper.selectOne(queryWrapper);
                if (existing == null) {
                    Likes likes = new Likes();
                    likes.setUserId(userId);
                    likes.setBizId(bizId);
                    likes.setMessageId(messageId);
                    likesMapper.insert(likes);
                }
            } else if ("REMOVE".equals(message.getOperationType())) {
                likesMapper.delete(queryWrapper);
            }
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理点赞操作消息异常: {}", e.getMessage(), e);
            // 执行NACK操作，拒绝消息并重新入队
            channel.basicNack(deliveryTag, false, true);
            // 重新抛出异常，以便事务回滚
            throw e;
        }
    }

    @SneakyThrows
    @Transactional(rollbackFor = RuntimeException.class)
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(name = LIKE_COUNT_QUEUE, durable = "true", arguments = {
                            @Argument(name = "x-dead-letter-exchange", value = DLX_EXCHANGE), // 设置死信交换机
                            @Argument(name = "x-dead-letter-routing-key", value = DLX_ROUTING_KEY),
                            @Argument(name = "x-message-ttl", value = "300000", type = "java.lang.Long")// 设置死信路由键
                    }),
                    exchange = @Exchange(name = LIKE_COUNT_EXCHANGE),
                    key = LIKE_COUNT_ROUTING_KEY),
            ackMode = "MANUAL")
    public void handleLikeCountMessage(LikeCountMessage message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            Integer bizId = message.getBizId();
            Long messageId = message.getMessageId();
            Long delta = message.getDelta();
            //数据库同步
            likeCountsMapper.upsertLikeCount(bizId, messageId, delta);
            Long count = likeCountsMapper.queryLikeCount(bizId, messageId);
            //视频同步
            if (bizId.equals(VIDEO_BIZ)) {
                videoMapper.updateLikesById(messageId, count);
                //评论同步
            } else if (bizId.equals(COMMENT_BIZ)) {
                commentMapper.updateLikesById(messageId, count);
            }
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理点赞计数消息异常: {}", e.getMessage(), e);
            // 执行NACK操作，拒绝消息并重新入队
            channel.basicNack(deliveryTag, false, true);
            // 重新抛出异常，以便事务回滚
            throw e;
        }
    }

    @SneakyThrows
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(name = COMMENT_CACHE_UPDATE_QUEUE, durable = "true", arguments = {
                            @Argument(name = "x-dead-letter-exchange", value = DLX_EXCHANGE), // 设置死信交换机
                            @Argument(name = "x-dead-letter-routing-key", value = DLX_ROUTING_KEY),
                            @Argument(name = "x-message-ttl", value = "300000", type = "java.lang.Long")// 设置死信路由键
                    }),
                    exchange = @Exchange(name = COMMENT_CACHE_UPDATE_EXCHANGE),
                    key = COMMENT_CACHE_UPDATE_ROUTING_KEY),
            ackMode = "MANUAL")
    public void handleCommentCache(CommentSelectMessage message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        List<CommentVO> commentVOS = message.getCommentVOS();
        Long videoId = message.getVideoId();
        boolean isRootComment = message.isRootComment();
        String key = isRootComment ? COMMENT_ROOT_VIDEO_KEY : COMMENT_CHILD_VIDEO_KEY + videoId;
        pipelineZSET(key + videoId, commentVOS);
        channel.basicAck(deliveryTag, false);
    }

    @SneakyThrows
    @Transactional(rollbackFor = RuntimeException.class, propagation = Propagation.REQUIRES_NEW)
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(name = COMMENT_CACHE_DELETE_QUEUE, durable = "true", arguments = {
                            @Argument(name = "x-dead-letter-exchange", value = DLX_EXCHANGE), // 设置死信交换机
                            @Argument(name = "x-dead-letter-routing-key", value = DLX_ROUTING_KEY),
                            @Argument(name = "x-message-ttl", value = "300000", type = "java.lang.Long")// 设置死信路由键
                    }),
                    exchange = @Exchange(name = COMMENT_CACHE_DELETE_EXCHANGE),
                    key = COMMENT_CACHE_DELETE_ROUTING_KEY),
            ackMode = "MANUAL")
    public void handleCommentCacheDelete(CommentDeleteMessage message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        CommentVO commentVO = message.getCommentVO();
        Long videoId = message.getVideoId();
        boolean isRootComment = message.isRootComment();
        Long toCommentId = message.getToCommentId();
        //删除一个评论，要删除所有指向他的所有评论
        String key = isRootComment ? COMMENT_ROOT_VIDEO_KEY : COMMENT_CHILD_VIDEO_KEY + videoId;
        //删除这个评论
        stringRedisTemplate.opsForZSet().remove(key, commentVO);
        localCache.cleanUp();
        //再删除这个数据库所有与他关联的评论
        if (toCommentId != null) {
            Db.removeById(toCommentId);
        }
        channel.basicAck(deliveryTag, false);
    }

    /**
     * 因为Redis本来就很快，所以将这个操作使用MQ有点画蛇添足了。。。
     * @param message
     * @param channel
     * @param deliveryTag
     */
    @SneakyThrows
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(name = LIKE_SYNC_REDIS_QUEUE, durable = "true", arguments = {
                            @Argument(name = "x-dead-letter-exchange", value = DLX_EXCHANGE), // 设置死信交换机
                            @Argument(name = "x-dead-letter-routing-key", value = DLX_ROUTING_KEY),
                            @Argument(name = "x-message-ttl", value = "300000", type = "java.lang.Long")// 设置死信路由键
                    }),
                    exchange = @Exchange(name = LIKE_SYNC_REDIS_EXCHANGE),
                    key = LIKE_SYNC_REDIS_ROUTING_KEY),
            ackMode = "MANUAL")
    public void likeLocal2Redis(LikeLocal2Redis message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        String userId = message.getUserId();
        boolean isLiked = message.isLiked();
        Integer bizId = message.getBizId();
        Long messageId = message.getMessageId();
        String likeSetKey = LIKE_SET_KEY + bizId + ":" + messageId;
        String likeCountKey = LIKE_COUNT_KEY + bizId + ":" + messageId;
        String luaScript = isLiked ?
                "redis.call('srem', KEYS[1], ARGV[1]); " +
                        "redis.call('decr', KEYS[2]);" :
                "redis.call('sadd', KEYS[1], ARGV[1]); " +
                        "redis.call('incr', KEYS[2]);";
        stringRedisTemplate.execute(
                new DefaultRedisScript<>(luaScript, Long.class),
                Arrays.asList(likeSetKey, likeCountKey),
                userId
        );
        // 发送MQ
        if (isLiked) {
            mqProducer.sendLikeOperationMessage("REMOVE", userId, bizId, messageId);
            mqProducer.sendLikeCountMessage(bizId, messageId, -1L);
        } else {
            mqProducer.sendLikeOperationMessage("ADD", userId, bizId, messageId);
            mqProducer.sendLikeCountMessage(bizId, messageId, 1L);
        }
        channel.basicAck(deliveryTag, false);
    }

    private void pipelineZSET(String key, List<CommentVO> commentVOS) {
        if (CollectionUtil.isEmpty(commentVOS)) {
            // 空页占位：添加一个特殊 member 占位
            stringRedisTemplate.opsForZSet().add(key, "EMPTY_PAGE", 0);
            stringRedisTemplate.expire(key, 5, TimeUnit.MINUTES);
            return;
        }
        RedisSerializer<String> stringSerializer = stringRedisTemplate.getStringSerializer();
        stringRedisTemplate.executePipelined((RedisCallback<String>) connection -> {
            for (CommentVO commentVO : commentVOS) {
                double score = calculateHotScore(commentVO);
                connection.zSetCommands().zAdd(
                        stringSerializer.serialize(key),
                        score,
                        stringSerializer.serialize(JSONUtil.toJsonStr(commentVO)));
            }
            return null;
        }, stringSerializer);
        stringRedisTemplate.expire(key, 30, TimeUnit.MINUTES);
    }


}
